package com.my.reactor;

import org.reactivestreams.Subscription;
import reactor.core.publisher.*;
import reactor.core.scheduler.Schedulers;

import java.io.IOException;
import java.time.Duration;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

/**
 * @author zhupanlin
 * @version 1.0
 * @description: TODO
 * @date 2024/2/7 17:31
 */
public class FluxDemo {


    public static void main(String[] args) throws IOException {
        new FluxDemo().thread();
        
        System.in.read();
    }
    
    public void thread(){
        // 响应式编程：全异步、消息、事件回调
        // 默认还是咏当前线程，生成整个流、发布流、流操作
        Flux.range(1, 10)
                .publishOn(Schedulers.single()) // 在哪个线程池吧这个流的数据和操作执行
                .log()
                .map(integer -> integer + 10)
                .log()
                .subscribe();
        
        // publishOn：改变发布者所在的线程池
        // subscribeOn：改变订阅者所在的线程池
        
        // 调度器 线程池
        Schedulers.immediate();// 无执行上下文，当前线程运行所有操作
        Schedulers.single();// 使用固定的一个单线程
        Schedulers.boundedElastic();// 有界、弹性调度；不是无线扩充的线程池；线程池中有 10 * CPU核心个线程
        Schedulers.fromExecutor(new ThreadPoolExecutor(4, 8, 60, TimeUnit.SECONDS, new LinkedBlockingDeque<>(1000)));
        
    }
    
    // 自定义流中元素自定义规则
    public void handle(){
        Flux.range(1, 10)
                .handle((value, sink) -> {
                    System.out.println("拿到的值：" + value); 
                    String user = "张三：" + value;
                    // 可以向下发送的通道
                    sink.next(user);
                }).log()
                .subscribe();
    }
    
    public void create(){
        //异步环境下
        Flux<Object> flux = Flux.create(fluxSink -> {
            MyListener myListener = new MyListener(fluxSink);
            for (int i = 0; i < 100; i++) {
                myListener.online("张" + i);
            }
        });
        flux.log()
                .subscribe();

    }
    
    class MyListener{
        FluxSink<Object> sink;
        public MyListener(FluxSink<Object> sink){
            this.sink = sink;
        }
        // 用户登录、触发online监听
        public void online(String username){
            System.out.println("用户登录了：" + username);
            sink.next(username); // 传入用户
        }
    }
    
    // 编程方式创建序列
    // Sink 接收器、水槽、通道
    // Source：数据源   Sink：接收端
    public void generate(){
        Flux<Object> flux = Flux.generate(() -> 0,  // state初始值
                (state, sink) -> {
                // 传递数据，可能会抛出Unchecked异常
                // 0 - 10
                    if (state > 10){
                        // 发送完成
                        sink.complete();
                    }
                    sink.next(state);
                    // 返回新的state的值
                    return state + 1;
        });
        
        flux.log()
                .subscribe();

    }
    
    public static void subscribe(String[] args) {
        Flux<String> flux = Flux.range(1, 7)
                .map(i -> {
                    System.out.println("元素-" + i);
                    return "哈哈-" + i;
                });
        flux.subscribe();
    }

    public static void fluxMono(String[] args) {
        Flux.concat(Flux.just(1, 2, 3), Flux.just(7, 8, 9))
                .subscribe(System.out::println);
        
        Flux.range(1, 7)
                //.log() // 日志
                .filter(i -> i > 3) // 挑出大于3的元素
                .log()
                .map(i -> "haha-" + i)
                //.log()
                .subscribe(System.out::println);
    }

    public static void doOnXxx(String[] args) {
        Flux.just(1, 2, 3, 4, 5, 6, 7, 0, 5, 6)
                .map(integer -> 10 / integer)
                .doOnError(throwable -> {
                    System.out.println("数据库已经保存了异常" + throwable);
                })
                .doOnNext(integer -> System.out.println("元素已到达：" + integer))
                .subscribe(System.out::println);
    }

    // Mono<Integer>：只有一个Integer
    // Flux<Integer>：有多个Integer
    public static void fluxDoOn(String[] args) {

        /*Mono<Integer> just = Mono.just(1);
        just.subscribe(System.out::println);*/

        // 空流 有一个信号，此时代表完成信号
        // 下面的操作符，操作的是上面的流
        // 事件感知：当流发生什么事的时候，触发一个回调；doOnXxx
        // 系统调用提前定义好的钩子函数（Hook 钩子函数）
        Flux<Integer> flux = Flux.range(1, 7)
                .delayElements(Duration.ofSeconds(1))
                .doOnComplete(() -> {
                    System.out.println("流结束了...");
                }).doOnCancel(() -> {
                    System.out.println("流已被取消");
                }).doOnError(throwable -> {
                    System.out.println("流出错了");
                }).doOnNext(integer -> {
                    System.out.println("doOnNext --> " + integer);
                });
        
        flux.subscribe(new BaseSubscriber<Integer>() {
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                System.out.println("订阅成功" + subscription);
                request(1);
            }

            @Override
            protected void hookOnNext(Integer value) {
                System.out.println("下一个元素：" + value);
                if (value < 5){
                    request(1);
                    if (value == 3){
                        int i = 10 / 0;
                    }
                }else {
                    cancel();
                }
            }

            @Override
            protected void hookOnComplete() {
                System.out.println("订阅结束");
            }

            @Override
            protected void hookOnError(Throwable throwable) {
                System.out.println("数据流异常：" + throwable);
            }

            @Override
            protected void hookOnCancel() {
                System.out.println("数据流被取消");
            }

            @Override
            protected void hookFinally(SignalType type) {
                // 相当于try-catch-finally中的finally
                System.out.println("结束了" + type);
            }
        });
        //flux.subscribe(System.out::println);


        try {
            System.in.read();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }


    }


    // 测试FLux
    public static void flux(String[] args) {
        // Mono：0|1个元素的流
        // Flux：N个元素的流
        // 发布者发布数据流：源头

        // 1.多元素的流
        Flux<Integer> just = Flux.just(1, 2, 3, 4, 5);

        // 流不消费就没用 不占内存：消费 == 订阅
        just.subscribe(System.out::println);
        // 一个数据流可以有很多消费者
        just.subscribe(e -> System.out.println("e2 = " + e));

        // 对于每个消费者来说流都是一样的；广播模式

        System.out.println("==============");
        // 每秒产生一个从0开始的递增数字
        Flux<Long> interval = Flux.interval(Duration.ofSeconds(1));
        interval.subscribe(System.out::println);

        try {
            Thread.sleep(6000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

}
